package com.qianbao.finance.sync.modules.elasticsearch.service.impl;

import com.google.common.collect.Maps;
import com.qianbao.finance.sync.modules.elasticsearch.service.ElasticsearchService;
import com.qianbao.finance.sync.modules.elasticsearch.util.JsonUtil;
import org.apache.http.HttpEntity;
import org.apache.http.entity.ContentType;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.http.util.EntityUtils;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;

@Service
public class ElasticsearchRestServiceImpl implements ElasticsearchService {
    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchRestServiceImpl.class);

    @Resource
    private RestHighLevelClient restHighLevelClient;

    @Resource
    private RestClient restClient;

    @Override
    public void insertById(String index, String type, String id, Map<String, Object> dataMap) {
        IndexRequest indexRequest = new IndexRequest(index, type, id);
        restHighLevelClient.indexAsync(indexRequest.source(dataMap), new ActionListener<IndexResponse>(){
            @Override
            public void onResponse(IndexResponse indexResponse) {
                logger.debug(indexResponse.toString());
            }

            @Override
            public void onFailure(Exception e) {
                logger.error("创建索引index[{}],type[{}],id[{}]失败", index, type, id, e);
            }
        } );
    }

    @Override
    public void upsertById(String index, String type, String id, Map<String, Object> dataMap) {
        UpdateRequest updateRequest = new UpdateRequest(index, type, id);
        restHighLevelClient.updateAsync(updateRequest.upsert(dataMap), new ActionListener<UpdateResponse>() {
            @Override
            public void onResponse(UpdateResponse updateResponse) {
                logger.debug(updateResponse.toString());
            }

            @Override
            public void onFailure(Exception e) {
                logger.error("更新索引失败index[{}],type[{}],id[{}]", index, type, id);
            }
        });
    }

    @Override
    public void batchInsertById(String index, String type, Map<String, Map<String, Object>> idDataMap) {
        BulkRequest bulkRequest = new BulkRequest();
        bulkRequest.timeout("5m");
        idDataMap.forEach((id, dataMap)->bulkRequest.add(new IndexRequest(index, type, id).source(dataMap)));
        try {
            BulkResponse bulkItemResponses = restHighLevelClient.bulk(bulkRequest);
            if(bulkItemResponses.hasFailures()){
                bulkItemResponses.forEach(bulkItemResponse -> {
                    if(bulkItemResponse.isFailed()){
                        BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                        logger.error("elasticsearch批量插入部分错误, index:[{}], type:[{}], id:[{}], status:[{}], cause:[{}], msg:[{}]" , failure.getIndex(), failure.getType(), failure.getId(), failure.getStatus(), failure.getCause(), failure.getMessage());
                    }
                });
            }
        } catch (IOException e) {
            logger.error("elasticsearch批量插入错误, index=" + index + ", type=" + type, e);
        }
/*        restHighLevelClient.bulkAsync(bulkRequest, new ActionListener<BulkResponse>() {
            @Override
            public void onResponse(BulkResponse bulkItemResponses) {
                if(bulkItemResponses.hasFailures()){
                    bulkItemResponses.forEach(bulkItemResponse -> {
                        if(bulkItemResponse.isFailed()){
                            BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                            logger.error("elasticsearch批量插入部分错误, index:[{}], type:[{}], id:[{}], status:[{}], cause:[{}], msg:[{}]" , failure.getIndex(), failure.getType(), failure.getId(), failure.getStatus(), failure.getCause(), failure.getMessage());
                        }
                    });
                }
            }

            @Override
            public void onFailure(Exception e) {
                logger.error("elasticsearch批量插入错误, index=" + index + ", type=" + type, e);
            }
        });*/
    }

    @Override
    public void update(String index, String type, String id, Map<String, Object> dataMap) {
        UpdateRequest updateRequest = new UpdateRequest(index, type, id);
        restHighLevelClient.updateAsync(updateRequest.doc(dataMap), new ActionListener<UpdateResponse>() {
            @Override
            public void onResponse(UpdateResponse updateResponse) {
                logger.debug(updateRequest.toString());
            }

            @Override
            public void onFailure(Exception e) {
                logger.error("更新索引index[{}],type[{}],id[{}]失败", index, type, id, e);
            }
        });
    }

    @Override
    public void deleteById(String index, String type, String id) {
        DeleteRequest deleteRequest = new DeleteRequest(index, type, id);
        restHighLevelClient.deleteAsync(deleteRequest, new ActionListener<DeleteResponse>() {
            @Override
            public void onResponse(DeleteResponse deleteResponse) {
                logger.debug(deleteResponse.toString());
            }

            @Override
            public void onFailure(Exception e) {
                logger.error("删除索引index[{}],type[{}],id[{}]失败", index, type, id, e);
            }
        });
    }

    @Override
    public void createMapping(String index, String type, Map<String, Map<String, Map<String, String>>> mappingMap) {
        Map<String, Map<String, Map<String, Map<String, Map<String, String>>>>> entityMap = Maps.newHashMap();
        Map<String, Map<String, Map<String, Map<String, String>>>> typeMap = Maps.newHashMap();
        entityMap.put("mappings", typeMap);
        typeMap.put(type, mappingMap);
        HttpEntity entity = new NStringEntity(JsonUtil.toJson(entityMap).toString(), ContentType.APPLICATION_JSON);
        try {
            Response response = restClient.performRequest("PUT", "/"+index, Collections.emptyMap(), entity);
        } catch (IOException e) {
            logger.error("upsert mapping exception for index:[{}], type:[{}]", index, type, e);
        }
    }

    @Override
    public void deleteIndex(String index) {
        try {
            Response response = restClient.performRequest("DELETE", "/"+index);
        } catch (IOException e) {
            logger.error("delete index exception for index:[{}]", index, e);
        }
    }

    public long count(String index, String type) {
        try{
            Response response = restClient.performRequest("GET", "/"+index+"/"+type+"/_count");
            Long count = Long.parseLong(JsonUtil.toMap(EntityUtils.toString(response.getEntity())).get("count").toString());
            logger.debug("count index [{}]-->[{}]", index, count);
            return count;
        }catch (IOException e) {
            logger.error("count index exception for index:[{}]", index, e);
        }
        return 0L;
    }
}
